package com.uhealin;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.Coprocessor;
import org.apache.hadoop.hbase.CoprocessorEnvironment;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.client.coprocessor.Batch;
import org.apache.hadoop.hbase.coprocessor.CoprocessorException;
import org.apache.hadoop.hbase.coprocessor.CoprocessorService;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.ipc.CoprocessorRpcUtils.BlockingRpcCallback;
import org.apache.hadoop.hbase.regionserver.InternalScanner;
import org.apache.hadoop.hbase.shaded.io.netty.handler.codec.http2.Http2Connection;
import org.apache.hadoop.hbase.shaded.protobuf.ResponseConverter;
import org.apache.hadoop.hbase.util.Bytes;

import com.google.protobuf.RpcCallback;
import com.google.protobuf.RpcController;
import com.google.protobuf.Service;
import com.uhealin.Sum.SumRequest;
import com.uhealin.Sum.SumResponse;
import com.uhealin.Sum.SumService;

public class SumEndpoint extends SumService implements Coprocessor,CoprocessorService{
	

	private RegionCoprocessorEnvironment env;
	
	@Override
	public void getSum(RpcController controller, SumRequest request, RpcCallback<SumResponse> done) {
		// TODO Auto-generated method stub

		 Scan scan=new Scan();
		 byte[] family=Bytes.toBytes(request.getFamily());
		 scan.addFamily(family);
		 scan.addColumn(family, Bytes.toBytes(request.getColunm()));
		 SumResponse response=null;
		 InternalScanner scanner=null;
		 try {
			  scanner=env.getRegion().getScanner(scan) ;
			  List<Cell> cells=new ArrayList<Cell>();
			  boolean hasMore=false;
			  long sum=0L;
			  do {
				  
				  hasMore=scanner.next(cells);
				  for(Cell cell :cells) {
					  sum += Long.parseLong(new String(CellUtil.cloneValue(cell)));
					  
				  }
				  cells.clear();
			  }while(hasMore);
			  response=SumResponse.newBuilder().setSum(sum).build();
			  
			  
		 }catch (IOException e) {
			// TODO: handle exception
			 ResponseConverter.setControllerException(controller, e);
		}finally {
			
			try {
				scanner.close();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			
		}
		 done.run(response);
		
	}


	public Service getService() {
		// TODO Auto-generated method stub
		return this;
	}

	public void start(CoprocessorEnvironment env) throws IOException {
		// TODO Auto-generated method stub
		if (env instanceof RegionCoprocessorEnvironment) {
			this.env=(RegionCoprocessorEnvironment)env;
		}else {
			throw new CoprocessorException("no load region");
		}
	}

	

	
	public static void main(String[] args) throws Throwable {
		// TODO Auto-generated method stub

		  Configuration conf=HBaseConfiguration.create();
		  conf.set("hbase.zookeeper.quorum", "localhost");
		  Connection conn=ConnectionFactory.createConnection(conf);
		  
		  Table table= conn.getTable(TableName.valueOf("sum"));
		  
		  long h=0L;
		  
		  final SumRequest request=SumRequest.newBuilder().setFamily("d").setColunm("b").build();
		  
		  Map<byte[],Long>  re=table.coprocessorService(SumService.class,null,null,new Batch.Call<SumService, Long>() {

			@Override
			public Long call(SumService instance) throws IOException {
				// TODO Auto-generated method stub
				BlockingRpcCallback<SumResponse> calllback=new BlockingRpcCallback<SumResponse>();
				instance.getSum(null, request, calllback);
				SumResponse response=calllback.get();
				return response.hasSum()?response.getSum():0L;
			}
		
			  
		  });
		  
		  for(Long sum:re.values()) {
			  h+=sum;
			  
		  }
		  
		  System.out.println("sum ="+h);
		  
		  
		  table.close();
		  conn.close();
		  
	}


}
